package p.ithorns.framework.redis.stream;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.Executor;

/**
 * @author liuhaitao
 * @date 2022/12/9 17:15
 */
public class RedisStreamFactory {

    private final static Logger log = LoggerFactory.getLogger(RedisStreamFactory.class);

    private final RedisConnectionFactory factory;

    private final RedisStreamTemplate<?> template;


    public RedisStreamFactory(RedisConnectionFactory factory,
                              RedisStreamTemplate<?> template) {
        this.factory = factory;
        this.template = template;
    }

    /**
     * 创建Subscription
     *
     * @param container -Stream容器
     * @param listener  -需要注册的StreamListener
     * @return Subscription
     */
    public <T> Subscription createSubscription(StreamMessageListenerContainer<String, Record<String, T>> container,
                                               RedisStreamListener<T> listener) {
        String stream = listener.getStream();
        initStream(stream);

        String group = listener.getGroup();
        boolean isGroupConsumer = listener.isGroupConsumer();
        StreamOffset<String> streamOffset = isGroupConsumer
                ? StreamOffset.create(stream, ReadOffset.lastConsumed())
                : StreamOffset.latest(stream);

        // 初始化的是StreamReadRequestBuilder
        StreamMessageListenerContainer.StreamReadRequestBuilder<String> builder =
                StreamMessageListenerContainer.StreamReadRequest.builder(streamOffset)
                        // 异常后不cancel, cancel了就无法继续接收消息了
                        // 而且这里要写在前面, 因为返回的是StreamReadRequest, 参考后面container.register的注释
                        // 可视情况配置处理，具体翻源码
                        .cancelOnError(t -> {
                            // log.warn("RedisStream Error ->: {}", t.getMessage());
                            return false;
                        });

        // 如果是消费组
        if (isGroupConsumer) {
            initGroup(stream, group);
            String groupName = Optional.ofNullable(listener.getGroupName()).orElse(group);
            Consumer consumer = Consumer.from(group, groupName);
            // 注意：这里变成了ConsumerStreamReadRequestBuilder：构建ConsumerStreamReadRequest，
            // {@link org.springframework.data.redis.stream.StreamPollTask#createPollState}
            // PollState.standalone()产生了一个空的Consumer对象(Optional.empty()),
            // 导致RedisStreamListener配置的Group和Consumer没有起到作用
            // 所以builder要覆盖, 否则默认的还是StreamReadRequestBuilder
            builder = builder.consumer(consumer).autoAcknowledge(listener.isAutoAck());
        }

        // 注册Subscription
        // ConsumerStreamReadRequest和StreamReadRequest是有区别的，
        // StreamReadRequest构建出来的无法触发组消费，要使用ConsumerStreamReadRequest;
        // 所以在builder的方法调用顺序上要注意
        return container.register(builder.build(), listener);
    }


    /**
     * 创建Stream容器
     *
     * @param targetType   -目标(消息)对象类型
     * @param executor     -执行器
     * @param errorHandler -异常处理器
     * @param <T>          目标(消息)类型
     * @return StreamMessageListenerContainer
     */
    public <T> StreamMessageListenerContainer createContainer(Class<T> targetType,
                                                              Executor executor,
                                                              ErrorHandler errorHandler
    ) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptionsBuilder builder
                = StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                .batchSize(100)
                .pollTimeout(Duration.ofSeconds(5))
                // 先指定Mapper, 否则会被targetType操作覆盖RedisTemplate中的RedisSerializer;
                .objectMapper(template.getHashMapper())
                // Stream Key的序列化方式
                .keySerializer(template.getKeySerializer())
                // Stream后方字段的 key 的序列化方式
                .hashKeySerializer(template.getKeySerializer())
                // Stream后方字段的 value 的序列化方式
                .hashValueSerializer(template.getHashValueSerializer());

        if (null != targetType && !targetType.isAssignableFrom(Map.class)) {
            // 指定具体类型，Listener的Record类型也必须是这个
            builder.targetType(targetType);
            // 关于原始类型问题: DefaultStreamOperations 会判定反序列化的类型，如果是原始类型，就不会走自定义HashMapper
            // 详情请翻看源码：DefaultStreamOperations: #line-69{isSimpleType(targetType)}
        }

        if (null != executor) {
            builder.executor(executor);
        }

        if (null != errorHandler) {
            builder.errorHandler(errorHandler);
        }


        StreamMessageListenerContainer.StreamMessageListenerContainerOptions options = builder.build();
        return StreamMessageListenerContainer.create(factory, options);
    }


    public void initStream(String key) {
        Assert.hasText(key, "Stream key must not be null or empty.");
        //判断key是否存在，如果不存在则创建
        if (!hasKey(key)) {
            HashMap hashMap = new HashMap();
            hashMap.put("k", 1);
            RecordId recordId = template.getStreamOps().add(key, hashMap);
            //将初始化的值删除掉
            if (null != recordId) {
                delField(key, recordId.getValue());
            }
        }
    }


    public void initGroup(String key, String groupName) {
        StreamOperations<String, Object, Object> opsStream = template.opsForStream();
        StreamInfo.XInfoGroups groups = opsStream.groups(key);
        boolean exist = groups.stream().anyMatch(group -> group.groupName().equals(groupName));
        if (!exist) {
            opsStream.createGroup(key, groupName);
        }
    }

    private void delField(String key, String value) {
        template.opsForStream().delete(key, value);
    }

    /**
     * 用来判断key是否存在
     */
    private boolean hasKey(String key) {
        if (key == null) {
            return false;
        } else {
            return Boolean.TRUE.equals(template.hasKey(key));
        }
    }

}